#include "epoll_loop.h"
#include "task.h"
#include <cerrno>
#include <chrono>
#include <coroutine>
#include <cstring>
#include <iostream>
#include <map>
#include <memory>
#include <queue>
#include <span>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <tuple>
#include <unistd.h>
#include <unordered_map>
#include <unordered_set>
#include <utility>

namespace cobt {
EpollLoop::EpollLoop() { m_epfd = epoll_create(1); }
EpollLoop::EpollLoop(EpollLoop &&that) {
  m_epfd = that.m_epfd;
  that.m_epfd = -1;
}
EpollLoop &EpollLoop::operator=(EpollLoop &&that) {
  m_epfd = that.m_epfd;
  that.m_epfd = -1;
  return *this;
}
EpollLoop::~EpollLoop() {
  if (m_epfd != -1) {
    close(m_epfd);
  }
}
void EpollLoop::add_event(epoll_event *event, int fd) {
  if (epoll_ctl(m_epfd, EPOLL_CTL_ADD, fd, event) < 0) {
    if (errno == EEXIST) {
      epoll_ctl(m_epfd, EPOLL_CTL_MOD, fd, event);
    } else {
      std::cout << "add_event error: " << strerror(errno) << std::endl;
    }
  }
}
void EpollLoop::remove_event(int fd) {
  epoll_ctl(m_epfd, EPOLL_CTL_DEL, fd, nullptr);
}
void EpollLoop::run() {
  auto now = std::chrono::steady_clock::now();
  for (auto it = m_timer.begin(); it != m_timer.end();) {
    if (now < it->first) {
      break;
    }
    auto cb = std::move(it->second);
    it = m_timer.erase(it);
    cb();
  }
  while (!m_coroutines.empty()) {
    auto handle = std::move(m_coroutines.front());
    m_coroutines.pop();
    handle.resume();
  }
  if (m_max_events == 0) {
    return;
  }
  int timeout = -1;
  auto dur = get_next_timer();
  if (dur != std::chrono::seconds::max()) {
    timeout =
        std::chrono::duration_cast<std::chrono::milliseconds>(dur).count();
  }
  int nums = epoll_wait(m_epfd, m_events, COBT_EPOLL_LOOP_MAX_EVENTS, timeout);
  for (int i = 0; i < nums; i++) {
    auto awaiter = static_cast<EventAwaiter *>(m_events[i].data.ptr);
    m_coroutines.push(awaiter->m_pre);
  }
}

void EpollLoop::enqueue(std::coroutine_handle<> coroutine) {
  m_coroutines.push(coroutine);
}

bool EpollLoop::empty() const noexcept {
  return m_coroutines.empty() && m_max_events == 0;
}

void EpollLoop::add_timer(std::chrono::steady_clock::time_point time,
                          std::function<void()> func) {
  m_timer.emplace(time, std::move(func));
}

std::chrono::steady_clock::duration EpollLoop::get_next_timer() const {
  if (m_timer.empty()) {
    return std::chrono::seconds::max();
  }
  return m_timer.begin()->first - std::chrono::steady_clock::now();
}
size_t EpollLoop::get_queue_size() const noexcept {
  return m_coroutines.size();
}

EventAwaiter::EventAwaiter(EpollLoop &loop, int fd, uint32_t events, int id)
    : m_loop(loop), m_id(id), m_fd(fd) {
  m_event.events = events;
  m_event.data.ptr = this;
  ++loop.m_max_events;
}

bool EventAwaiter::await_ready() const noexcept { return false; }
void EventAwaiter::await_suspend(std::coroutine_handle<> handle) {
  m_pre = handle;
  m_loop.add_event(&m_event, m_fd);
}
void EventAwaiter::await_resume() const noexcept {}

EventAwaiter::~EventAwaiter() {
  m_loop.remove_event(m_fd);
  --m_loop.m_max_events;
}

using TcpEvent =
    std::tuple<bool, std::chrono::steady_clock::time_point, EventAwaiter *>;

struct CompareByTimePoint {
  bool operator()(const TcpEvent &a, const TcpEvent &b) const {
    return std::get<1>(a) < std::get<1>(b);
  }
};

Task<int> read_async(int fd, std::span<char> buf,
                     std::chrono::milliseconds timeout) {
  auto awaiter =
      std::make_shared<EventAwaiter>(g_loop, fd, EPOLLIN | EPOLLET, 1);
  // g_awaiter_set.insert(awaiter.get());
  // if (timeout != std::chrono::milliseconds::max()) {
  //   set_timeout(timeout, [fd, awaiter]() {
  //     if (!g_awaiter_set.contains(awaiter.get())) {
  //       return;
  //     }
  //     g_awaiter_set.erase(awaiter.get());
  //     awaiter->m_pre.resume();
  //   });
  // }
  int res = 0;
  while (true) {
    // if (!g_awaiter_set.contains(awaiter.get())) {
    //   std::cout << "read timeout 111" << std::endl;
    //   co_return 0;
    // }
    res = read(fd, buf.data(), buf.size());
    if (res >= 0 || errno != EAGAIN) {
      break;
    }
    co_await *awaiter;
  }
  // g_awaiter_set.erase(awaiter.get());
  co_return res;
}

Task<int> write_async(int fd, std::span<char> buf,
                      std::chrono::milliseconds timeout) {
  auto awaiter =
      std::make_shared<EventAwaiter>(g_loop, fd, EPOLLOUT | EPOLLET, 1);
  // g_awaiter_set.insert(awaiter.get());
  // if (timeout != std::chrono::milliseconds::max()) {
  //   set_timeout(timeout, [fd, awaiter]() {
  //     if (!g_awaiter_set.contains(awaiter.get())) {
  //       return;
  //     }
  //     g_awaiter_set.erase(awaiter.get());
  //     awaiter->m_pre.resume();
  //   });
  // }
  int res;
  while (true) {
    // if (!g_awaiter_set.contains(awaiter.get())) {
    //   std::cout << "write timeout 111" << std::endl;
    //   co_return 0;
    // }
    res = write(fd, buf.data(), buf.size());
    if (res >= 0 || errno != EAGAIN) {
      break;
    }
    co_await *awaiter;
  }

  // g_awaiter_set.erase(awaiter.get());
  co_return res;
}

Task<int> accept_async(int fd) {
  co_await EventAwaiter(g_loop, fd, EPOLLIN, 2);
  co_return accept(fd, nullptr, nullptr);
}

Task<int> connect_async(int fd, const SocketAddress &addr,
                        std::chrono::milliseconds timeout) {
  int flags = fcntl(fd, F_GETFL, 0);
  fcntl(fd, F_SETFL, flags | O_NONBLOCK);
  while (true) {
    int res = connect(fd, addr.get_addr(), addr.get_addr_len()) < 0;
    if (res == 0) {
      co_return 0;
    }
    if (errno != EINPROGRESS) {
      // std::cout << strerror(errno) << std::endl;
      co_return res;
    }
    co_await EventAwaiter(g_loop, fd, EPOLLOUT | EPOLLET, 3);
  }

  co_return 0;
}

void set_timeout(std::chrono::steady_clock::time_point expire_time,
                 std::function<void()> cb) {
  g_loop.add_timer(expire_time, cb);
}
void set_timeout(std::chrono::milliseconds timeout, std::function<void()> cb) {
  auto stop_end = std::chrono::steady_clock::now() + timeout;
  g_loop.add_timer(stop_end, cb);
}

} // namespace cobt